package juc.scenarios.compute;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

/**
 * Fork/Join框架并发计算演示
 * 演示大数相加的并发计算，展示分治算法的并发实现
 */
public class ForkJoinComputeDemo {

    /**
     * 大数相加任务
     */
    static class BigNumberSumTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 10000; // 任务分割阈值
        private final long[] array;
        private final int start;
        private final int end;

        public BigNumberSumTask(long[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Long compute() {
            // 如果任务足够小，直接计算
            if (end - start <= THRESHOLD) {
                long sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                return sum;
            }

            // 任务太大，分割成两个子任务
            int mid = start + (end - start) / 2;
            BigNumberSumTask leftTask = new BigNumberSumTask(array, start, mid);
            BigNumberSumTask rightTask = new BigNumberSumTask(array, mid, end);

            // 异步执行子任务
            leftTask.fork();
            rightTask.fork();

            // 等待子任务完成并合并结果
            long leftResult = leftTask.join();
            long rightResult = rightTask.join();

            return leftResult + rightResult;
        }
    }

    /**
     * 数组排序任务
     */
    static class ArraySortTask extends RecursiveAction {
        private static final int THRESHOLD = 1000;
        private final long[] array;
        private final int start;
        private final int end;

        public ArraySortTask(long[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        @Override
        protected void compute() {
            if (end - start <= THRESHOLD) {
                // 小数组使用直接排序
                Arrays.sort(array, start, end);
            } else {
                // 大数组分割排序
                int mid = start + (end - start) / 2;
                ArraySortTask leftTask = new ArraySortTask(array, start, mid);
                ArraySortTask rightTask = new ArraySortTask(array, mid, end);

                // 执行并等待完成
                invokeAll(leftTask, rightTask);

                // 合并排序结果
                merge(array, start, mid, end);
            }
        }

        private void merge(long[] array, int start, int mid, int end) {
            long[] temp = new long[end - start];
            int i = start, j = mid, k = 0;

            while (i < mid && j < end) {
                if (array[i] <= array[j]) {
                    temp[k++] = array[i++];
                } else {
                    temp[k++] = array[j++];
                }
            }

            while (i < mid) {
                temp[k++] = array[i++];
            }

            while (j < end) {
                temp[k++] = array[j++];
            }

            System.arraycopy(temp, 0, array, start, temp.length);
        }
    }

    /**
     * 并行计算演示
     */
    public static void demonstrateParallelComputing() {
        System.out.println("=== Fork/Join框架并发计算演示 ===\n");

        // 生成测试数据
        int arraySize = 1_000_000;
        long[] data = generateRandomArray(arraySize);
        System.out.println("生成测试数据：" + arraySize + " 个随机数");

        // 创建ForkJoinPool
        ForkJoinPool forkJoinPool = new ForkJoinPool(
                Runtime.getRuntime().availableProcessors(),
                ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                null,
                true
        );

        try {
            // 1. 并行求和演示
            demonstrateSumComputing(forkJoinPool, data);

            // 2. 并行排序演示
            demonstrateSorting(forkJoinPool, data);

            // 3. 性能对比
            performanceComparison(data);

            // 4. 工作窃取演示
            demonstrateWorkStealing(forkJoinPool);

        } finally {
            forkJoinPool.shutdown();
            try {
                if (!forkJoinPool.awaitTermination(60, TimeUnit.SECONDS)) {
                    forkJoinPool.shutdownNow();
                }
            } catch (InterruptedException e) {
                forkJoinPool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 并行求和演示
     */
    private static void demonstrateSumComputing(ForkJoinPool pool, long[] data) {
        System.out.println("\n--- 并行求和演示 ---");

        // Fork/Join并行计算
        BigNumberSumTask task = new BigNumberSumTask(data, 0, data.length);

        long startTime = System.nanoTime();
        long parallelResult = pool.invoke(task);
        long parallelTime = System.nanoTime() - startTime;

        // 串行计算对比
        startTime = System.nanoTime();
        long sequentialResult = 0;
        for (long value : data) {
            sequentialResult += value;
        }
        long sequentialTime = System.nanoTime() - startTime;

        System.out.println("Fork/Join并行计算结果: " + parallelResult);
        System.out.println("串行计算结果: " + sequentialResult);
        System.out.println("结果验证: " + (parallelResult == sequentialResult ? "✓ 正确" : "✗ 错误"));
        System.out.println("并行计算耗时: " + String.format("%.2f ms", parallelTime / 1_000_000.0));
        System.out.println("串行计算耗时: " + String.format("%.2f ms", sequentialTime / 1_000_000.0));
        System.out.println("性能提升: " + String.format("%.2f%%",
                (sequentialTime - parallelTime) * 100.0 / sequentialTime));

        // 显示线程池状态
        System.out.println("活跃线程数: " + pool.getActiveThreadCount());
        System.out.println("并行度: " + pool.getParallelism());
        System.out.println("工作线程数: " + pool.getPoolSize());
    }

    /**
     * 并行排序演示
     */
    private static void demonstrateSorting(ForkJoinPool pool, long[] originalData) {
        System.out.println("\n--- 并行排序演示 ---");

        // 复制数据用于排序
        long[] dataToSort = Arrays.copyOf(originalData, originalData.length);

        ArraySortTask sortTask = new ArraySortTask(dataToSort, 0, dataToSort.length);

        long startTime = System.nanoTime();
        pool.invoke(sortTask);
        long parallelTime = System.nanoTime() - startTime;

        // 验证排序结果
        boolean isSorted = true;
        for (int i = 1; i < dataToSort.length; i++) {
            if (dataToSort[i - 1] > dataToSort[i]) {
                isSorted = false;
                break;
            }
        }

        System.out.println("并行排序结果验证: " + (isSorted ? "✓ 正确" : "✗ 错误"));
        System.out.println("并行排序耗时: " + String.format("%.2f ms", parallelTime / 1_000_000.0));
        System.out.println("排序数据量: " + dataToSort.length + " 个元素");
    }

    /**
     * 性能对比
     */
    private static void performanceComparison(long[] data) {
        System.out.println("\n--- 性能对比分析 ---");

        int[] testSizes = {10000, 50000, 100000, 500000, 1000000};

        for (int size : testSizes) {
            if (size > data.length) break;

            long[] testData = Arrays.copyOf(data, size);

            // 串行计算
            long startTime = System.nanoTime();
            long sequentialSum = 0;
            for (int i = 0; i < size; i++) {
                sequentialSum += testData[i];
            }
            long sequentialTime = System.nanoTime() - startTime;

            // 并行计算
            ForkJoinPool pool = new ForkJoinPool();
            BigNumberSumTask task = new BigNumberSumTask(testData, 0, testData.length);

            startTime = System.nanoTime();
            long parallelSum = pool.invoke(task);
            long parallelTime = System.nanoTime() - startTime;

            pool.shutdown();

            double speedup = (double) sequentialTime / parallelTime;

            System.out.println(String.format("数据量: %,d | 串行: %.2f ms | 并行: %.2f ms | 加速比: %.2fx | 效率: %.1f%%",
                    size,
                    sequentialTime / 1_000_000.0,
                    parallelTime / 1_000_000.0,
                    speedup,
                    (speedup / Runtime.getRuntime().availableProcessors()) * 100));
        }
    }

    /**
     * 工作窃取演示
     */
    private static void demonstrateWorkStealing(ForkJoinPool pool) {
        System.out.println("\n--- 工作窃取算法演示 ---");

        class WorkStealingTask extends RecursiveTask<Integer> {
            private final int workload;
            private final String taskName;

            public WorkStealingTask(int workload, String taskName) {
                this.workload = workload;
                this.taskName = taskName;
            }

            @Override
            protected Integer compute() {
                String threadName = Thread.currentThread().getName();
                System.out.println(String.format("任务 %s 在线程 %s 上执行 (工作量: %d)",
                        taskName, threadName, workload));

                if (workload > 100) {
                    // 分割任务
                    int mid = workload / 2;
                    WorkStealingTask left = new WorkStealingTask(mid, taskName + "-L");
                    WorkStealingTask right = new WorkStealingTask(workload - mid, taskName + "-R");

                    left.fork();
                    right.fork();

                    return left.join() + right.join();
                } else {
                    // 模拟工作
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return workload;
                }
            }
        }

        // 创建不平衡的工作负载
        WorkStealingTask task1 = new WorkStealingTask(500, "Task-1");
        WorkStealingTask task2 = new WorkStealingTask(300, "Task-2");
        WorkStealingTask task3 = new WorkStealingTask(200, "Task-3");

        pool.invoke(task1);
        pool.invoke(task2);
        pool.invoke(task3);

        System.out.println("工作窃取演示完成，观察线程间的任务分配");
    }

    /**
     * 生成随机数组
     */
    private static long[] generateRandomArray(int size) {
        Random random = new Random(42); // 固定种子确保可重复性
        long[] array = new long[size];
        for (int i = 0; i < size; i++) {
            array[i] = random.nextInt(1000);
        }
        return array;
    }

    public static void main(String[] args) {
        demonstrateParallelComputing();
    }
}